package geektime.spring.data.reactive.redisdemo;

import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class DemoSubsriber
        implements Subscriber<Integer> {

    static final Logger LOGGER = LoggerFactory.getLogger(DemoSubsriber.class);

    @Override
    public void onSubscribe(Subscription s) {
        LOGGER.info("request max");
        s.request(Integer.MAX_VALUE);
    }

    @Override
    public void onNext(Integer integer) {
        LOGGER.info("get data:{}", integer);

    }

    @Override
    public void onError(Throwable t) {

    }

    @Override
    public void onComplete() {
        LOGGER.info("onComplete");
    }


    @Test
    public void testDoOnMethods() {
        Flux.range(1, 3)
                //                .log()
                .doOnSubscribe(e -> LOGGER.info("doOnSubscribe:{}", e))
                .doOnEach(e -> LOGGER.info("doOnEach:{}", e))
                .doOnError(e -> LOGGER.info("doOnError:{}", e))
                .doOnNext(e -> LOGGER.info("doOnNext:{}", e))
                .doOnRequest(e -> LOGGER.info("doOnRequest:{}", e))
                .doOnTerminate(new Runnable() {
                    @Override
                    public void run() {
                        LOGGER.info("doOnTerminate");
                    }
                })
                .doOnCancel(new Runnable() {
                    @Override
                    //                    public void run() {
                    //                    LOGGER.info("doOnCancel");
                    //                  }
                    public void run() {
                        LOGGER.info("操作被取消");
                    }
                })
                .doOnComplete(new Runnable() {
                    @Override
                    public void run() {
                        LOGGER.info("doOnComplete");
                    }
                })
                .subscribe(new DemoSubsriber());

    }

    @Test
    public void testDoOnMethods1() throws InterruptedException {
        Flux.range(1, 3)
                .publishOn(Schedulers.single())
                .doOnSubscribe(e -> LOGGER.info("doOnSubscribe:{}", e))
                .doOnComplete(() -> LOGGER.info("doOnComplete"))
                .subscribe(c -> LOGGER.info("subscribe-->Insert{} records", c), e -> LOGGER.info("subscribe-->doOnSubscribe:{}", e), new Runnable() {
                    @Override
                    public void run() {
                        LOGGER.info("subscribe-->doOnTerminate");
                    }
                });
        Thread.sleep(900000L);

    }

    //  @Test
    //  public void testConnectableFluxRefConnect() throws InterruptedException {
    //
    //    Flux<Long> source = Flux.interval(Duration.ofMillis(500))
    //        .doOnSubscribe(s -> System.out.println("上遊收到訂閱"))
    //        .doOnCancel(() -> System.out.println("上遊發布者斷開連接doOnCancel"));
    //
    //    Flux<Long> refCounted = source.publish().refCount(2, Duration.ofSeconds(2));
    //
    //    System.out.println("第一個訂閱者訂閱");
    //    Disposable sub1 = refCounted.subscribe(l -> System.out.println("sub1: " + l));
    //
    //    TimeUnit.SECONDS.sleep(1);
    //    System.out.println("第二個訂閱者訂閱");
    //    Disposable sub2 = refCounted.subscribe(l -> System.out.println("sub2: " + l));
    //
    //    TimeUnit.SECONDS.sleep(1);
    //    System.out.println("第一個訂閱者取消訂閱");
    //    sub1.dispose();
    //
    //    TimeUnit.SECONDS.sleep(1);
    //    System.out.println("第二個訂閱者取消訂閱");
    //    sub2.dispose();
    //
    //    TimeUnit.SECONDS.sleep(1);
    //    System.out.println("第三個訂閱者訂閱");
    //    Disposable sub3 = refCounted.subscribe(l -> System.out.println("sub3: " + l));
    //
    //    TimeUnit.SECONDS.sleep(1);
    //    System.out.println("第三個訂閱者取消訂閱");
    //    sub3.dispose();
    //
    //    TimeUnit.SECONDS.sleep(3);
    //    System.out.println("第四個訂閱者訂閱");
    //    Disposable sub4 = refCounted.subscribe(l -> System.out.println("sub4: " + l));
    //    TimeUnit.SECONDS.sleep(1);
    //    System.out.println("第五個訂閱者訂閱");
    //    Disposable sub5 = refCounted.subscribe(l -> System.out.println("sub5: " + l));
    //    TimeUnit.SECONDS.sleep(2);
    //  }
}